package com.suxin.kafka.thread;

import com.alibaba.fastjson.JSON;
import com.suxin.kafka.thread.domain.OrderDto;
import com.suxin.kafka.thread.service.OrderService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.AbstractConsumerSeekAware;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;

/**
 * @author Tang
 * @classname OrderKafkaListener
 * @description [  ]
 * @date 2022/2/15 22:18
 */
public class OrderKafkaListener extends AbstractConsumerSeekAware {

    OrderService orderService;

    KafkaConsumerPool kafkaConsumerPool;

    public OrderKafkaListener() {
        KafkaConsumerConfig<OrderDto> consumerConfig = new KafkaConsumerConfig<>();
        consumerConfig.setBizService(orderService::handler);
        consumerConfig.setBizName("order-thread");
        consumerConfig.setConcurrentSize(3);
        kafkaConsumerPool = new KafkaConsumerPool(consumerConfig);
    }

    @KafkaListener(topics = {"${kafka.order.topic}"}, containerFactory = "")
    public void consumerMsg(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment) {

        records.forEach(consumerRecord -> {
            OrderDto orderDto = JSON.parseObject(JSON.toJSONString(consumerRecord.value()), OrderDto.class);
            this.kafkaConsumerPool.addTask(orderDto.getMsgId(), orderDto);
        });

        acknowledgment.acknowledge();
    }


}